package com.jml.mapreduce.自定义使用inputFormat;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

/**
 * 这个RecordReader负责将整个文件读成(转化成)一组key，value对
 * key就是文件路径+名字
 * value就是文件内容
 *
 * 一个RecordReader就对应一个文件
 */
public class MyRecordReader extends RecordReader<Text, BytesWritable>  {

    //默认false表示没读过
    private boolean isRead = false;

    private Text key = new Text();
    private BytesWritable value = new BytesWritable();

    FSDataInputStream inputStream;
    FileSplit fs;

    /**
     *初始化方法
     * @param split：里面有文件路径信息
     * @param context
     */
    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        //类型转换,FileInputFormat返回的split是FileSplit
        fs = (FileSplit)split;
        //开流
        FileSystem fileSystem = FileSystem.get(context.getConfiguration());
        //这个路径一定是输入的一部分
        inputStream = fileSystem.open(fs.getPath());
    }

    /**
     *判断是否有下一个kv
     */
    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        //这个文件只有一个kv值,一次读就读完了
        //但还没有读取的时候
        if(!isRead){
            //没有读取就进行读取文件,并且填充kv
            //key:文件路径+文件名
            key.set(fs.getPath().toString());
            //value:文件内容
            byte[] buffer = new byte[(int) fs.getLength()];
            //文件读取进buffer
            inputStream.read(buffer);
            value.set(buffer,0,buffer.length);
            isRead = true;
            return true;
        }else{
            return false;
        }
    }

    /**
     *得到当前的key
     */
    @Override
    public Text getCurrentKey() throws IOException, InterruptedException {
        return key;
    }

    /**
     *得到当前value
     */
    @Override
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {
        return value;
    }

    /**
     *当前RecordReader在数据中的进度(百分比)
     */
    @Override
    public float getProgress() throws IOException, InterruptedException {
        return isRead?1:0;
    }

    /**
     *关闭方法,一般用来关闭资源
     */
    @Override
    public void close() throws IOException {
        //关流
        IOUtils.closeStream(inputStream);
    }
}
